package fun.tan90.easy.binlog.plugin.rocketmq.config;


import com.github.benmanes.caffeine.cache.CacheLoader;
import fun.tan90.easy.binlog.plugin.common.constants.DamiEventConstants;
import fun.tan90.easy.binlog.plugin.common.model.OutputParam;
import fun.tan90.easy.binlog.plugin.rocketmq.model.RocketmqClientParams;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.noear.dami.Dami;
import org.springframework.lang.NonNull;
import org.springframework.lang.Nullable;

import java.util.Objects;

/**
 * description 动态连接mqtt
 * company 铁人科技
 *
 * @author Li YuLong
 * @date 2023/8/17 10:56
 */
@Slf4j
public class RocketmqClientCacheLoader implements CacheLoader<String, DefaultMQProducer> {

    @Override
    public @Nullable DefaultMQProducer load(@NonNull String outputCode) throws Exception {
        OutputParam outputParam = Dami.<String, OutputParam>bus().sendAndRequest(DamiEventConstants.OUTPUT_PARAM_TOPIC, outputCode);
        if (Objects.isNull(outputParam)) {
            return null;
        }
        RocketmqClientParams rocketmqClientParams = (RocketmqClientParams) outputParam;

        /**
         * producerGroup（必需）：生产者组名，用于逻辑上标识属于同一组的生产者实例。在一个生产者组中，每个实例可以并行地发送消息。例如：producerGroup=MyProducerGroup。
         * namesrvAddr（必需）：RocketMQ 的命名服务器地址列表，用于获取 Broker 的路由信息。格式为 "host1:port1;host2:port2;..."。通过该参数，生产者能够找到正确的 Broker 来发送消息。例如：namesrvAddr=192.168.0.1:9876;192.168.0.2:9876。
         * instanceName：生产者实例名称，用于在同一个生产者组内标识不同的实例。如果不指定，默认值为 "DEFAULT"。
         * sendMsgTimeout：发送消息的超时时间（毫秒）。如果在超时时间内没有收到消息确认，则认为发送失败。默认值为 3000 毫秒。
         * retryTimesWhenSendFailed：发送失败时的重试次数。当发生可重试错误时，生产者将尝试重新发送消息。默认值为 2。
         * retryTimesWhenSendAsyncFailed：异步发送失败时的重试次数。当异步发送消息失败时，生产者将尝试重新发送消息。默认值为 2。
         * maxMessageSize：消息的最大大小（字节）。默认值为 4MB。如果消息的大小超过该值，则会被拒绝发送。
         * compressionType：发送消息时的压缩类型。支持的压缩类型有 "NONE"、"ZIP" 和 "GZIP"。默认值为 "NONE"（不压缩）。
         * retryAnotherBrokerWhenNotStoreOK：如果发送消息到当前 Broker 失败，是否选择另一个 Broker 发送。默认值为 true。
         * vipChannelEnabled：是否启用 VIP 通道。VIP 通道是一种高优先级的网络通道，用于提供更低的网络延迟。默认值为 true。
         */
        DefaultMQProducer producer = new DefaultMQProducer(rocketmqClientParams.getNamespace(), rocketmqClientParams.getProducerGroup());
        producer.setNamesrvAddr(rocketmqClientParams.getNamesrvAddr());
        producer.setSendMsgTimeout(rocketmqClientParams.getSendMsgTimeout());
        producer.setRetryAnotherBrokerWhenNotStoreOK(rocketmqClientParams.isRetryAnotherBrokerWhenNotStoreOk());
        producer.setRetryTimesWhenSendAsyncFailed(rocketmqClientParams.getRetryTimesWhenSendAsyncFailed());
        producer.start();
        return producer;
    }
}